/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.datashark.plugin.task.api;

import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import com.datashark.common.constants.Constants;
import com.datashark.common.exception.BaseException;
import com.datashark.common.utils.HttpUtils;
import com.datashark.common.utils.JSONUtils;
import com.datashark.common.utils.KerberosHttpClient;
import com.datashark.common.utils.PropertyUtils;
import com.datashark.plugin.task.api.model.TaskResponse;
import com.datashark.plugin.task.api.shell.IShellInterceptorBuilder;
import com.datashark.plugin.task.api.shell.ShellInterceptorBuilderFactory;
import com.datashark.plugin.task.api.utils.LogUtils;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import lombok.extern.slf4j.Slf4j;

import static com.datashark.common.constants.Constants.*;
import static com.datashark.common.constants.Constants.YARN_APPLICATION_STATUS_ADDRESS;
import static com.datashark.plugin.task.api.TaskConstants.EXIT_CODE_KILL;

@Slf4j
public abstract class AbstractYarnTask extends AbstractRemoteTask {

    private static final int MAX_RETRY_ATTEMPTS = 3;

    private ShellCommandExecutor shellCommandExecutor;

    public AbstractYarnTask(TaskExecutionContext taskRequest) {
        super(taskRequest);
        this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle, taskRequest);
    }

    @Override
    public void submitApplication() throws TaskException {
        try {
            IShellInterceptorBuilder shellActuatorBuilder =
                    ShellInterceptorBuilderFactory.newBuilder()
                            .properties(getProperties())
                            // todo: do we need to move the replace to subclass?
                            .appendScript(getScript().replaceAll("\\r\\n", System.lineSeparator()));
            // SHELL task exit code
            TaskResponse response = shellCommandExecutor.run(shellActuatorBuilder, null);
            setExitStatusCode(response.getExitStatusCode());
            setAppIds(String.join(TaskConstants.COMMA, getApplicationIds()));
            setProcessId(response.getProcessId());
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            log.info("The current yarn task has been interrupted", ex);
            setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
            throw new TaskException("The current yarn task has been interrupted", ex);
        } catch (Exception e) {
            log.error("yarn process failure", e);
            exitStatusCode = -1;
            throw new TaskException("Execute task failed", e);
        }
    }

    @Override
    public void trackApplicationStatus() throws TaskException {
        if (StringUtils.isEmpty(appIds)) {
            return;
        }


        List<String> appIdList = Arrays.asList(appIds.split(","));
        boolean continueTracking = true;

        while (continueTracking) {
            Map<String, YarnState> yarnStateMap = new HashMap<>();
            for (String appId : appIdList) {
                if (StringUtils.isEmpty(appId)) {
                    continue;
                }

                boolean hadoopSecurityAuthStartupState =
                        PropertyUtils.getBoolean(HADOOP_SECURITY_AUTHENTICATION_STARTUP_STATE, false);
                String yarnStateJson = fetchYarnStateJsonWithRetry(appId, hadoopSecurityAuthStartupState);
                if (StringUtils.isNotEmpty(yarnStateJson)) {
                    String appJson = JSONUtils.getNodeString(yarnStateJson, "app");
                    YarnTask yarnTask = JSONUtils.parseObject(appJson, YarnTask.class);
                    log.info("yarnTask : {}", yarnTask);
                    yarnStateMap.put(yarnTask.getId(), YarnState.of(yarnTask.getState()));
                }
            }

            YarnState yarnTaskOverallStatus = YarnTaskStatusChecker.getYarnTaskOverallStatus(yarnStateMap);
            if (yarnTaskOverallStatus.isFinalState()) {
                handleFinalState(yarnTaskOverallStatus);
                continueTracking = false;
            } else {
                try {
                    TimeUnit.MILLISECONDS.sleep(SLEEP_TIME_MILLIS * 10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private String fetchYarnStateJsonWithRetry(String appId,
                                               boolean hadoopSecurityAuthStartupState) throws TaskException {
        int retryCount = 0;
        while (retryCount < MAX_RETRY_ATTEMPTS) {
            try {
                return fetchYarnStateJson(appId, hadoopSecurityAuthStartupState);
            } catch (Exception e) {
                retryCount++;
                log.error("Failed to fetch or parse Yarn state for appId: {}. Attempt: {}/{}",
                        appId, retryCount, MAX_RETRY_ATTEMPTS, e);

                if (retryCount >= MAX_RETRY_ATTEMPTS) {
                    throw new TaskException("Failed to fetch Yarn state after "
                            + MAX_RETRY_ATTEMPTS + " attempts for appId: " + appId, e);
                }

                try {
                    TimeUnit.MILLISECONDS.sleep(SLEEP_TIME_MILLIS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(ie);
                }
            }
        }
        return null;
    }

    private void handleFinalState(YarnState yarnState) {
        switch (yarnState) {
            case FINISHED:
                setExitStatusCode(EXIT_CODE_SUCCESS);
                break;
            case KILLED:
                setExitStatusCode(EXIT_CODE_KILL);
                break;
            default:
                setExitStatusCode(EXIT_CODE_FAILURE);
                break;
        }
    }

    private String fetchYarnStateJson(String appId, boolean hadoopSecurityAuthStartupState) throws Exception {
        return hadoopSecurityAuthStartupState
                ? KerberosHttpClient.get(getApplicationUrl(appId))
                : HttpUtils.get(getApplicationUrl(appId));
    }


    static class YarnTaskStatusChecker {

        public static YarnState getYarnTaskOverallStatus(Map<String, YarnState> yarnTaskMap) {
            // 检查是否有任何任务处于 FAILED 或 KILLED 状态
            boolean hasKilled = yarnTaskMap.values().stream()
                    .anyMatch(state -> state == YarnState.KILLED);

            if (hasKilled) {
                return YarnState.KILLED;
            }

            // 检查是否有任何任务处于 FAILED 或 KILLED 状态
            boolean hasFailed = yarnTaskMap.values().stream()
                    .anyMatch(state -> state == YarnState.FAILED);

            if (hasFailed) {
                return YarnState.FAILED;
            }


            // 检查是否所有任务都处于 FINISHED 状态
            boolean allFINISHED = yarnTaskMap.values().stream()
                    .allMatch(state -> state == YarnState.FINISHED);

            if (allFINISHED) {
                return YarnState.FINISHED;
            }

            // 检查是否有任何任务处于 RUNNING 状态
            boolean hasRunning = yarnTaskMap.values().stream()
                    .anyMatch(state -> state == YarnState.RUNNING);

            if (hasRunning) {
                return YarnState.RUNNING;
            }

            // 检查是否有任何任务处于提交中状态
            boolean hasSubmitting = yarnTaskMap.values().stream()
                    .anyMatch(state -> state == YarnState.NEW || state == YarnState.NEW_SAVING
                            || state == YarnState.SUBMITTED || state == YarnState.ACCEPTED);

            if (hasSubmitting) {
                return YarnState.SUBMITTING;
            }

            // 如果都不匹配，返回未知状态
            return YarnState.UNKNOWN;
        }
    }


    /**
     * cancel application
     *
     * @throws TaskException exception
     */
    @Override
    public void cancelApplication() throws TaskException {
        // cancel process
        try {
            shellCommandExecutor.cancelApplication();
        } catch (Exception e) {
            throw new TaskException("cancel application error", e);
        }
    }

    /**
     * get application ids
     *
     * @return
     * @throws TaskException
     */
    @Override
    public List<String> getApplicationIds() throws TaskException {
        // TODO 这里看common.properties中是否配置 appId.collect了，如果配置了走aop，否则走log
        return LogUtils.getAppIds(
                taskRequest.getLogPath(),
                taskRequest.getAppInfoPath(),
                PropertyUtils.getString(APPID_COLLECT, DEFAULT_COLLECT_WAY));
    }

    /** Get the script used to bootstrap the task */
    protected abstract String getScript();

    /** Get the properties of the task used to replace the placeholders in the script. */
    protected abstract Map<String, String> getProperties();

    @Data
    static class YarnTask {
        private String id;
        private String state;
    }

    private String getApplicationUrl(String applicationId) throws BaseException {

        String yarnResourceRmIds = PropertyUtils.getString(YARN_RESOURCEMANAGER_HA_RM_IDS);
        String yarnAppStatusAddress = PropertyUtils.getString(YARN_APPLICATION_STATUS_ADDRESS);
        String hadoopResourceManagerHttpAddressPort =
                PropertyUtils.getString(HADOOP_RESOURCE_MANAGER_HTTPADDRESS_PORT);

        String appUrl = StringUtils.isEmpty(yarnResourceRmIds) ?
                yarnAppStatusAddress :
                getAppAddress(yarnAppStatusAddress, yarnResourceRmIds);

        if (StringUtils.isBlank(appUrl)) {
            throw new BaseException("yarn application url generation failed");
        }

        log.info("yarn application url:{}", String.format(appUrl, hadoopResourceManagerHttpAddressPort, applicationId));
        return String.format(appUrl, hadoopResourceManagerHttpAddressPort, applicationId);
    }

    private static String getAppAddress(String appAddress, String rmHa) {

        String[] appAddressArr = appAddress.split(Constants.DOUBLE_SLASH);

        if (appAddressArr.length != 2) {
            return null;
        }

        String protocol = appAddressArr[0] + Constants.DOUBLE_SLASH;
        String[] pathSegments = appAddressArr[1].split(Constants.COLON);

        if (pathSegments.length != 2) {
            return null;
        }

        String end = Constants.COLON + pathSegments[1];

        // get active ResourceManager
        String activeRM = YarnHAAdminUtils.getActiveRMName(protocol, rmHa);

        if (StringUtils.isEmpty(activeRM)) {
            return null;
        }

        return protocol + activeRM + end;
    }

    /** yarn ha admin utils */
    private static final class YarnHAAdminUtils {

        /**
         * get active resourcemanager node
         *
         * @param protocol http protocol
         * @param rmIds yarn ha ids
         * @return yarn active node
         */
        public static String getActiveRMName(String protocol, String rmIds) {

            String hadoopResourceManagerHttpAddressPort =
                    PropertyUtils.getString(HADOOP_RESOURCE_MANAGER_HTTPADDRESS_PORT);

            String[] rmIdArr = rmIds.split(Constants.COMMA);

            String yarnUrl = protocol
                    + "%s:"
                    + hadoopResourceManagerHttpAddressPort
                    + "/ws/v1/cluster/info";
            try {
                /** send http get request to rm */
                for (String rmId : rmIdArr) {
                    String state = getRMState(String.format(yarnUrl, rmId));
                    if (Constants.HADOOP_RM_STATE_ACTIVE.equals(state)) {
                        return rmId;
                    }
                }

            } catch (Exception e) {
                log.error("get yarn ha application url failed", e);
            }
            return null;
        }

        /** get ResourceManager state */
        public static String getRMState(String url) {
            boolean hadoopSecurityAuthStartupState =
                    PropertyUtils.getBoolean(HADOOP_SECURITY_AUTHENTICATION_STARTUP_STATE, false);
            String retStr = Boolean.TRUE.equals(hadoopSecurityAuthStartupState)
                    ? KerberosHttpClient.get(url)
                    : HttpUtils.get(url);

            if (StringUtils.isEmpty(retStr)) {
                return null;
            }
            // to json
            ObjectNode jsonObject = JSONUtils.parseObject(retStr);

            // get ResourceManager state
            if (!jsonObject.has("clusterInfo")) {
                return null;
            }
            return jsonObject.get("clusterInfo").path("haState").asText();
        }
    }

    public enum YarnState {
        NEW,
        NEW_SAVING,
        SUBMITTED,
        ACCEPTED,
        RUNNING,
        FINISHED,
        FAILED,
        KILLED,
        SUBMITTING,
        UNKNOWN,
        ;

        // 将字符串转换为枚举
        public static YarnState of(String state) {
            try {
                return YarnState.valueOf(state);
            } catch (IllegalArgumentException | NullPointerException e) {
                // 如果字符串无效，则返回 null
                return null;
            }
        }

        /**
         * 任务结束
         * @return
         */
        public boolean isFinalState() {
            return this == FINISHED || this == FAILED || this == KILLED;
        }
    }
}